package com.atguigu.tingshu.search.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.extra.pinyin.PinyinUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.AlbumFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumInfoIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserFeignClient;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"all"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;

    @Autowired
    private AlbumFeignClient albumFeignClient;

    @Autowired
    private UserFeignClient userFeignClient;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 期望单例Bean但找到多个，造成无法注入
     * 方式1：@Autowired 注入先按类型注入，在按照Bean名称注入
     * 方式2：@Autowired + @Qualifier 指定BeanID完成注入
     * 方式3：@Resource 先按Bean名称注入，在按类型注入
     */
    //@Autowired
    //@Qualifier("threadPoolTaskExecutor")
    @Resource(name = "threadPoolTaskExecutor")
    private Executor threadPoolTaskExecutor;

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 将指定专辑保存到索引库
     *
     * @param albumId
     */
    @Override
    public void upperAlbum(Long albumId) {
        //1.创建专辑索引对象
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        //2.远程调用专辑服务查询专辑信息，封装专辑以及专辑标签列表
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            AlbumInfo albumInfo = albumFeignClient.getAlbumInfo(albumId).getData();
            Assert.notNull(albumInfo, "专辑{}不存在", albumId);
            //2.1 封装专辑信息
            BeanUtil.copyProperties(albumInfo, albumInfoIndex);

            //2.2 封装专辑标签列表
            List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
            if (CollUtil.isNotEmpty(albumAttributeValueVoList)) {
                List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueVoList
                        .stream()
                        .map(a -> BeanUtil.copyProperties(a, AttributeValueIndex.class))
                        .collect(Collectors.toList());
                albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
            }
            return albumInfo;
        }, threadPoolTaskExecutor);

        //3.远程调用专辑服务查询分类信息，封装1,2级分类ID
        CompletableFuture<Void> BaseCategoryViewCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            BaseCategoryView baseCategoryView = albumFeignClient.getCategoryView(albumInfo.getCategory3Id()).getData();
            Assert.notNull(baseCategoryView, "分类{}不存在", albumInfo.getCategory3Id());
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
        }, threadPoolTaskExecutor);

        //4.远程调用用户服务查询用户信息，封装主播名称
        CompletableFuture<Void> userInfoCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            UserInfoVo userInfoVo = userFeignClient.getUserInfoVo(albumInfo.getUserId()).getData();
            Assert.notNull(userInfoVo, "用户{}不存在", albumInfo.getUserId());
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        }, threadPoolTaskExecutor);

        //5.暂时随机为当前专辑生成四项统计数值，基于四项统计数值计算得出热度
        CompletableFuture<Void> statCompletableFuture = CompletableFuture.runAsync(() -> {
            int num1 = RandomUtil.randomInt(2000, 3000);
            int num2 = RandomUtil.randomInt(1000, 2000);
            int num3 = RandomUtil.randomInt(500, 1000);
            int num4 = RandomUtil.randomInt(200, 500);
            albumInfoIndex.setPlayStatNum(num1);
            albumInfoIndex.setSubscribeStatNum(num2);
            albumInfoIndex.setBuyStatNum(num3);
            albumInfoIndex.setCommentStatNum(num4);
            //基于四项统计数值计算得出热度 规则：播放量*0.1 + 订阅量*0.2 + 购买量*0.3 + 评论量*0.4
            BigDecimal hotScore = new BigDecimal(num1).multiply(new BigDecimal("0.1"))
                    .add(new BigDecimal(num2).multiply(new BigDecimal("0.2")))
                    .add(new BigDecimal(num3).multiply(new BigDecimal("0.3")))
                    .add(new BigDecimal(num4).multiply(new BigDecimal("0.4")));
            albumInfoIndex.setHotScore(hotScore.doubleValue());
        }, threadPoolTaskExecutor);

        //6.等待所有任务执行完成
        CompletableFuture.allOf(
                        albumInfoCompletableFuture,
                        BaseCategoryViewCompletableFuture,
                        userInfoCompletableFuture,
                        statCompletableFuture
                ).orTimeout(1, TimeUnit.SECONDS)
                .join();

        //7.保存专辑到索引库
        albumInfoIndexRepository.save(albumInfoIndex);

        //8.将专辑标题作为提示词文档存入提示词索引库
        this.saveSuggestIndex(albumInfoIndex);

        //9.将专辑ID存入布隆过滤器，用于解决缓存穿透问题
        RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumId);
    }

    /**
     * 删除专辑索引库中文档
     *
     * @param albumId
     */
    @Override
    public void lowerAlbum(Long albumId) {
        albumInfoIndexRepository.deleteById(albumId);
        suggestIndexRepository.deleteById(albumId.toString());
    }

    @Autowired
    private ElasticsearchClient elasticsearchClient;


    /**
     * 站内多条件检索
     * 整体采用传统写法，局部采用Lambda写法
     *
     * @param albumIndexQuery
     * @return
     */
    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {
        try {
            //一、构建检索请求对象
            SearchRequest searchRequest = this.buildDSL(albumIndexQuery);
            System.err.println("本次检索DSL,复制到Kibana验证：");
            System.err.println(searchRequest.toString());

            //二、执行检索
            SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);

            //三、解析结果返回
            return this.parseResult(searchResponse, albumIndexQuery);
        } catch (IOException e) {
            log.error("执行检索失败：{}", e.getMessage());
            throw new RuntimeException(e);
        }
    }

    private static final String INDEX_NAME = "albuminfo";

    /**
     * 构建完整检索请求对象，包含所有DSL请求体参数
     *
     * @param albumIndexQuery 条件
     * @return
     */
    @Override
    public SearchRequest buildDSL(AlbumIndexQuery albumIndexQuery) {
        //1.创建检索对象构建器对象
        SearchRequest.Builder builder = new SearchRequest.Builder();
        //2.设置索引库名称
        builder.index(INDEX_NAME);

        //3.逐项封装请求体参数
        //3.1 设置请求体参数"query"查询条件
        //3.1.1 创建bool查询组合三大查询条件
        BoolQuery.Builder allConditionBoolQueryBuilder = new BoolQuery.Builder();
        //3.1.2 处理关键字查询条件
        String keyword = albumIndexQuery.getKeyword();
        if (StringUtils.isNotBlank(keyword)) {
            allConditionBoolQueryBuilder.must(m -> m.match(m1 -> m1.field("albumTitle").query(keyword)));
        }

        //3.1.3 处理1,2,3级分类过滤条件
        Long category1Id = albumIndexQuery.getCategory1Id();
        if (category1Id != null) {
            allConditionBoolQueryBuilder.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)));
        }
        Long category2Id = albumIndexQuery.getCategory2Id();
        if (category2Id != null) {
            allConditionBoolQueryBuilder.filter(f -> f.term(t -> t.field("category2Id").value(category2Id)));
        }
        Long category3Id = albumIndexQuery.getCategory3Id();
        if (category3Id != null) {
            allConditionBoolQueryBuilder.filter(f -> f.term(t -> t.field("category3Id").value(category3Id)));
        }

        //3.1.4 处理多组标签过滤条件 标签id:标签值id
        List<String> attributeList = albumIndexQuery.getAttributeList();
        if (CollUtil.isNotEmpty(attributeList)) {
            for (String attribute : attributeList) {
                //每循环一次封装一个filter对应的Nested查询对象
                String[] split = attribute.split(":");
                if (split != null && split.length == 2) {
                    allConditionBoolQueryBuilder.filter(f -> f.nested(
                            n -> n.path("attributeValueIndexList")
                                    .query(q -> q.bool(b -> b.must(m -> m.term(t -> t.field("attributeValueIndexList.attributeId").value(split[0])))
                                            .must(m -> m.term(t -> t.field("attributeValueIndexList.valueId").value(split[1])))))
                    ));
                }
            }
        }

        builder.query(allConditionBoolQueryBuilder.build()._toQuery());
        //3.2 设置请求体参数"from","size"分页
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        int from = (pageNo - 1) * pageSize;
        builder.from(from).size(pageSize);
        //3.3 设置请求体参数"highlight"高亮
        if (StringUtils.isNotBlank(keyword)) {
            builder.highlight(h -> h.fields("albumTitle", h1 -> h1.preTags("<font style='color:red'>").postTags("</font>")));
        }
        //3.4 设置请求体参数"sort"排序  形式：高亮字段（1,2,3）：排序规则
        String order = albumIndexQuery.getOrder();
        if (StringUtils.isNotBlank(order)) {
            String[] split = order.split(":");
            if (split != null && split.length == 2) {
                //确定排序字段 以及 排序方向
                String orderField = "";
                switch (split[0]) {
                    case "1":
                        orderField = "hotScore";
                        break;
                    case "2":
                        orderField = "playStatNum";
                        break;
                    case "3":
                        orderField = "createTime";
                        break;
                }
                SortOrder sortOrder = split[1].equals("asc") ? SortOrder.Asc : SortOrder.Desc;
                String finalOrderField = orderField;
                builder.sort(s -> s.field(f -> f.field(finalOrderField).order(sortOrder)));
            }
        }
        //3.5 设置请求体参数"_source"指定字段响应
        builder.source(s -> s.filter(f -> f.excludes("attributeValueIndexList",
                "hotScore",
                "commentStatNum",
                "buyStatNum",
                "subscribeStatNum",
                "announcerName")));

        //4.基于检索对象构建器对象返回实际检索请求对象
        return builder.build();
    }

    /**
     * 解析ES返回检索结果
     *
     * @param searchResponse
     * @param albumIndexQuery
     * @return
     */
    @Override
    public AlbumSearchResponseVo parseResult(SearchResponse<AlbumInfoIndex> searchResponse, AlbumIndexQuery albumIndexQuery) {
        //1.创建响应VO对象
        AlbumSearchResponseVo vo = new AlbumSearchResponseVo();
        //2.封装VO中分页相关四项属性
        //2.1 从入参直接获取页码跟页大小
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        //2.2 从ES响应结果获取总记录数
        long total = searchResponse.hits().total().value();
        //2.3 计算总页数
        long totalPages = total % pageSize == 0 ? total / pageSize : total / pageSize + 1;
        vo.setPageNo(pageNo);
        vo.setPageSize(pageSize);
        vo.setTotal(total);
        vo.setTotalPages(totalPages);

        //3.封装VO中检索到专辑列表集合属性
        List<Hit<AlbumInfoIndex>> hits = searchResponse.hits().hits();
        if (CollUtil.isNotEmpty(hits)) {
            //采用Stream流将检索到文档类型从AlbumInfoIndex转为AlbumInfoIndexVo
            List<AlbumInfoIndexVo> list = hits.stream().map(hit -> {
                AlbumInfoIndex albumInfoIndex = hit.source();
                //处理高亮字段
                Map<String, List<String>> stringListMap = hit.highlight();
                if (CollUtil.isNotEmpty(stringListMap)) {
                    String hightLightText = stringListMap.get("albumTitle").get(0);
                    albumInfoIndex.setAlbumTitle(hightLightText);
                }
                return BeanUtil.copyProperties(albumInfoIndex, AlbumInfoIndexVo.class);
            }).collect(Collectors.toList());
            vo.setList(list);
        }
        //4.响应VO对象
        return vo;
    }

    /**
     * 检索七个置顶分类以及包含热度top6专辑列表
     *
     * @param category1Id
     * @return
     */
    @Override
    public List<Map<String, Object>> channel(Long category1Id) {
        try {
            //1.远程调用专辑服务，获取1级分类下7个置顶分类列表
            //1.1 发起远程调用
            List<BaseCategory3> baseCategory3List = albumFeignClient.findTopBaseCategory3(category1Id).getData();
            Assert.notNull(baseCategory3List, "没有查询到分类列表");
            //1.2 为了后续封装置顶分类热门专辑Map对象中 分类信息 ，将分类集合转为Map对象 key:分类ID value:分类对象
            Map<Long, BaseCategory3> category3Map = baseCategory3List
                    .stream()
                    .collect(Collectors.toMap(BaseCategory3::getId, c3 -> c3));

            //1.3 基于3级分类集合转为FieldValue封装每个三级分类ID
            List<FieldValue> fieldValueList =
                    baseCategory3List.stream().map(c3 -> FieldValue.of(c3.getId())).collect(Collectors.toList());

            //2.检索ES获取每个置顶分类及包含热度前六专辑列表
            SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(
                    s -> s.index(INDEX_NAME)
                            .query(q -> q.terms(t -> t.field("category3Id").terms(t1 -> t1.value(fieldValueList))))
                            .aggregations(
                                    "c3_agg", a -> a.terms(t -> t.field("category3Id").size(7))
                                            .aggregations(
                                                    "top6", a1 -> a1.topHits(t -> t.size(6).sort(s1 -> s1.field(f -> f.field("hotScore").order(SortOrder.Desc))).source(s1 -> s1.filter(f -> f.excludes("attributeValueIndexList",
                                                            "hotScore",
                                                            "commentStatNum",
                                                            "buyStatNum",
                                                            "subscribeStatNum",
                                                            "announcerName"))))
                                            )
                            )
                            .size(0),
                    AlbumInfoIndex.class);

            //3.解析ES检索结果，封装数据并返回
            Map<String, Aggregate> aggregations = searchResponse.aggregations();
            if (CollUtil.isNotEmpty(aggregations)) {
                //3.1 将聚合结果转为LongTermsAggregate获取聚合buckets桶数组
                List<LongTermsBucket> buckets = aggregations.get("c3_agg").lterms().buckets().array();
                //3.2 遍历buckets桶数组，每遍历一次产出置顶分类热门专辑Map对象
                List<Map<String, Object>> mapList = buckets.stream().map(bucket -> {
                    //3.2.1 创建置顶分类热门专辑Map对象
                    Map<String, Object> map = new HashMap<>();
                    //3.2.2 封装三级分类对象
                    long category3Id = bucket.key();
                    map.put("baseCategory3", category3Map.get(category3Id));
                    //3.2.3 封装对应热度最高top6专辑列表 将子聚合类型转为TopHitsAggregate
                    List<AlbumInfoIndex> top6List = bucket.aggregations()
                            .get("top6")
                            .topHits()
                            .hits()
                            .hits()
                            .stream()
                            .map(hit -> {
                                //采用FastJson将专辑JSON转为索引库文档对象
                                return JSON.parseObject(hit.source().toString(), AlbumInfoIndex.class);
                            }).collect(Collectors.toList());
                    map.put("list", top6List);
                    return map;
                }).collect(Collectors.toList());
                return mapList;
            }
            return null;
        } catch (IOException e) {
            log.error("ES检索异常：{}", e.getMessage());
            throw new RuntimeException(e);
        }
    }

    /**
     * 专辑标题作为提示词文档存入提示词索引库
     *
     * @param albumInfoIndex
     */
    @Override
    public void saveSuggestIndex(AlbumInfoIndex albumInfoIndex) {
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumInfoIndex.getId().toString());
        //原始内容
        String albumTitle = albumInfoIndex.getAlbumTitle();
        suggestIndex.setTitle(albumTitle);
        //用于汉字自动补全字段
        suggestIndex.setKeyword(new Completion(new String[]{albumTitle}));
        //用于汉语拼音自动补全字段
        String pinyin = PinyinUtil.getPinyin(albumTitle, "");
        suggestIndex.setKeywordPinyin(new Completion(new String[]{pinyin}));
        //用于汉语拼音首字母自动补全字段
        String firstLetter = PinyinUtil.getFirstLetter(albumTitle, "");
        suggestIndex.setKeywordSequence(new Completion(new String[]{firstLetter}));
        suggestIndexRepository.save(suggestIndex);
    }

    private static final String SUGGEST_INDEX = "suggestinfo";

    /**
     * 根据用户已录入字符展示提示词下拉列表（最多10个）
     *
     * @param keyword
     * @return
     */
    @Override
    public List<String> completeSuggest(String keyword) {
        try {
            //1.优先进行检索提示词索引库
            SearchResponse<SuggestIndex> searchResponse = elasticsearchClient.search(s -> s.index(SUGGEST_INDEX).suggest(
                    s1 -> s1.suggesters("keywordSuggest", suggest -> suggest.prefix(keyword).completion(c -> c.field("keyword").skipDuplicates(true)))
                            .suggesters("pinyinSuggest", suggest -> suggest.prefix(keyword).completion(c -> c.field("keywordPinyin").skipDuplicates(true)))
                            .suggesters("sequenceSuggest", suggest -> suggest.prefix(keyword).completion(c -> c.field("keywordSequence").skipDuplicates(true)))
            ), SuggestIndex.class);
            //2.解析自动补全结果，将提示词列表放入hashset集合中(去重)
            Set<String> hashSet = new HashSet<>();
            hashSet.addAll(this.parseSuggestResult(searchResponse, "keywordSuggest"));
            hashSet.addAll(this.parseSuggestResult(searchResponse, "pinyinSuggest"));
            hashSet.addAll(this.parseSuggestResult(searchResponse, "sequenceSuggest"));

            //3.如果上面结果数量小于10个，尝试采用全文检索补齐到10个
            if (hashSet.size() < 10) {
                SearchResponse<AlbumInfoIndex> search = elasticsearchClient.search(
                        s -> s.index(INDEX_NAME)
                                .query(q -> q.match(m -> m.field("albumTitle").query(keyword)))
                                .source(s1 -> s1.filter(f -> f.includes("albumTitle")))
                                .size(10), AlbumInfoIndex.class);
                List<Hit<AlbumInfoIndex>> hitList = search.hits().hits();
                if (CollUtil.isNotEmpty(hitList)) {
                    for (Hit<AlbumInfoIndex> hit : hitList) {
                        String albumTitle = hit.source().getAlbumTitle();
                        hashSet.add(albumTitle);
                        if (hashSet.size() >= 10) {
                            break;
                        }
                    }
                }
            }
            //4.将所有提示词列表返回
            if (hashSet.size() >= 10) {
                return new ArrayList<>(hashSet).subList(0, 10);
            }
            return new ArrayList<>(hashSet);
        } catch (IOException e) {
            log.error("建议词自动补全异常", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 解析自动补全建议结果
     *
     * @param searchResponse 检索响应结果对象
     * @param suggestName    建议器名称
     * @return
     */
    @Override
    public Collection<String> parseSuggestResult(SearchResponse<SuggestIndex> searchResponse, String suggestName) {
        List<String> list = new ArrayList<>();
        List<Suggestion<SuggestIndex>> suggestionList = searchResponse.suggest().get(suggestName);
        if (CollUtil.isNotEmpty(suggestionList)) {
            for (Suggestion<SuggestIndex> suggestion : suggestionList) {
                for (CompletionSuggestOption<SuggestIndex> option : suggestion.completion().options()) {
                    String title = option.source().getTitle();
                    list.add(title);
                }
            }
        }
        return list;
    }

    /**
     * 更新Redis小时榜数据
     *
     * @return
     */
    @Override
    public void updateLatelyAlbumRanking(Integer top) {
        try {
            //1.从Redis检索获取分类下五种排序维度专辑TOPN记录
            //1.1 远程调用专辑服务获取所有1级分类列表
            List<BaseCategory1> baseCategory1List = albumFeignClient.findAllCategory1().getData();
            Assert.notNull(baseCategory1List, "1级分类列表为空");


            //1.2 遍历1级分类
            for (BaseCategory1 baseCategory1 : baseCategory1List) {
                Long category1Id = baseCategory1.getId();
                //基于基于1级分类ID构建Redis中hash结构的Key
                String key = RedisConstant.RANKING_KEY_PREFIX + category1Id;
                //基于Key创建绑定hash操作对象
                BoundHashOperations<String, String, List<AlbumInfoIndex>> hashOps = redisTemplate.boundHashOps(key);
                //1.3 遍历5种排序维度
                String[] rankingDimensionArray
                        = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
                for (String dimension : rankingDimensionArray) {
                    //1.4 根据1级分类ID+排序字段检索ES
                    SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(s ->
                            s.index(INDEX_NAME)
                                    .query(q -> q.term(t -> t.field("category1Id").value(category1Id)))
                                    .sort(s1 -> s1.field(f -> f.field(dimension).order(SortOrder.Desc)))
                                    .size(top)
                                    .source(s1 -> s1.filter(f -> f.excludes("attributeValueIndexList",
                                            "hotScore",
                                            "commentStatNum",
                                            "buyStatNum",
                                            "subscribeStatNum",
                                            "announcerName"))), AlbumInfoIndex.class);
                    //2.解析ES检索结果 获取专辑TOPN记录
                    List<Hit<AlbumInfoIndex>> hits = searchResponse.hits().hits();
                    if (CollUtil.isNotEmpty(hits)) {
                        List<AlbumInfoIndex> topNList
                                = hits.stream().map(hit -> hit.source())
                                .collect(Collectors.toList());
                        //3.将TOPN记录存入Redis中hash小时排行中
                        //3.1 构建Hash排行榜中field 当前排序方式
                        String field = dimension;
                        //redisTemplate.opsForHash().put(key, field, topNList);
                        hashOps.put(field, topNList);
                    }
                }
            }
        } catch (IOException e) {
            log.error("更新Redis小时榜数据异常", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取不同分类下不同排行维度TOP20
     *
     * @param category1Id
     * @param dimension
     * @return
     */
    @Override
    public List<AlbumInfoIndexVo> getRankingList(Long category1Id, String dimension) {
        //基于基于1级分类ID构建Redis中hash结构的Key
        String key = RedisConstant.RANKING_KEY_PREFIX + category1Id;
        BoundHashOperations<String, String, List<AlbumInfoIndex>> hashOps = redisTemplate.boundHashOps(key);
        //构建Hash排行榜中field 当前排序方式
        String field = dimension;
        Boolean flag = hashOps.hasKey(field);
        if(flag){
            //List<AlbumInfoIndex> list = (List<AlbumInfoIndex>) redisTemplate.opsForHash().get(key, field);
            List<AlbumInfoIndex> list = hashOps.get(field);
            if (CollUtil.isNotEmpty(list)) {
                List<AlbumInfoIndexVo> albumInfoIndexVoList = list.stream()
                        .map(albumInfoIndex -> BeanUtil.copyProperties(albumInfoIndex, AlbumInfoIndexVo.class))
                        .collect(Collectors.toList());
                return albumInfoIndexVoList;
            }
        }
        return null;
    }


    /**
     #增量更新
     POST /albuminfo/_update/1
     {
     "script": {
     "source": "ctx._source.playStatNum += params.increment",
     "lang": "painless",
     "params": {
     "increment": 1
     }
     }
     }
     */
    /**
     * 更新专辑统计数值
     *
     * @param mqVo
     */
    @Override
    public void updateAlbumStat(TrackStatMqVo mqVo) {
        try {
            Long albumId = mqVo.getAlbumId();
            String incrementField = "";
            if (SystemConstant.TRACK_STAT_PLAY.equals(mqVo.getStatType())) {
                incrementField = "playStatNum";

            } else if (SystemConstant.TRACK_STAT_COMMENT.equals(mqVo.getStatType())) {
                incrementField = "commentStatNum";
            }
            String finalIncrementField = incrementField;
            elasticsearchClient.update(
                    u -> u.index(INDEX_NAME).id(albumId.toString())
                            .script(s -> s.inline(i -> i.source("ctx._source." + finalIncrementField + " += params.increment").lang("painless").params(Map.of("increment", JsonData.of(mqVo.getCount()))))),
                    AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
